package jp.co.worksap.ate.gomoku.hadoopai;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;

import jp.co.worksap.ate.gomoku.localai.GobanNode;
import jp.co.worksap.ate.gomoku.model.GoState;
import jp.co.worksap.ate.gomoku.model.Goban;

import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.SequenceFile;
import org.apache.hadoop.io.SequenceFile.CompressionType;
import org.apache.hadoop.io.SequenceFile.Writer;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.JobConf;

public class HadoopUtil {
	public static JobConf getBasicConf() throws IOException {
		JobConf conf = new JobConf();
		InputStream in = HadoopUtil.class.getClassLoader().getResourceAsStream(
				"hadoop-config.properties");
		Properties props = new Properties();
		props.load(in);
		conf.set("mapred.job.tracker", props.getProperty("mapred.job.tracker"));
		conf.set("fs.default.name", props.getProperty("fs.default.name"));
		conf.setJar(props.getProperty("jarpath"));
		return conf;
	}

	public static void createGomokuFolder(String folderName) throws IOException {
		Path outputPath = new Path("/gomoku/" + folderName + "/output");
		Path inputPath = new Path("/gomoku/" + folderName + "/input");
		JobConf conf = getBasicConf();
		FileSystem fs = FileSystem.get(conf);
		if (!fs.exists(inputPath)) {
			fs.mkdirs(inputPath);
		}
		if (fs.exists(outputPath)) {
			fs.delete(outputPath, true);
		}
	}

	public static void deleteGomokuFolder(String folderName) throws IOException {
		Path path = new Path("/gomoku/" + folderName);
		JobConf conf = getBasicConf();
		FileSystem fs = FileSystem.get(conf);
		if (fs.exists(path)) {
			fs.delete(path, true);
		}
	}

	public static void writeInputFile(String folderName, Goban goban,
			GoState goType, List<GobanNode> rootList, int level)
			throws IOException {
		JobConf conf = getBasicConf();
		FileSystem fs = FileSystem.get(conf);
		for (GobanNode rootNode : rootList) {
			GobanParam param = new GobanParam();
			param.setGoban(goban);
			param.setGoType(goType);
			param.setLevel(level);
			param.setX(rootNode.getX());
			param.setY(rootNode.getY());
			Text key = new Text(rootNode.getX() + "_" + rootNode.getY());
			Path inputFilePath = new Path("/gomoku/" + folderName
					+ "/input/input_" + key);
			Writer fwriter = SequenceFile.createWriter(fs, conf, inputFilePath,
					Text.class, GobanParam.class, CompressionType.NONE);
			fwriter.append(key, param);
			fwriter.close();
		}
	}

	public static Map<String, String> readOutputFile(String folderName)
			throws IOException {
		Path outputDirPath = new Path("/gomoku/" + folderName + "/output/");

		Map<String, String> map = new HashMap<String, String>();
		JobConf conf = getBasicConf();
		FileSystem fs = FileSystem.get(conf);
		FileStatus[] stats = fs.listStatus(outputDirPath);
		for (FileStatus stat : stats) {
			Path outputFilePath = stat.getPath();
			if (!fs.isFile(outputFilePath)) {
				continue;
			}
			FSDataInputStream in = fs.open(stat.getPath());
			BufferedReader freader = new BufferedReader(new InputStreamReader(
					in));
			String line;
			while ((line = freader.readLine()) != null) {
				String[] parts = line.split("\\s+");
				map.put(parts[0], parts[1]);
			}
			freader.close();
		}
		return map;
	}
}
